package cn.doitedu.javase.thread;

import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONObject;
import org.apache.commons.io.FileUtils;
import org.apache.commons.io.IOUtils;

import java.io.*;
import java.util.*;

/**
 * @Author: 深似海
 * @Site: <a href="www.51doit.com">多易教育</a>
 * @QQ: 657270652
 * @Date: 2024/2/2
 * @Desc: 学大数据，上多易教育
 * <p>
 * 用之前的json数据模板，生成一个10000行的文件
 * <p>
 * 读取上面生成的文件，全部读出来，放到一个List<String>中，一行一个元素
 * <p>
 * 把上面的list中的数据按照   余数 = event_id.hashcode() % 5
 * 余数等于0的数据，全部交给线程0处理
 * 余数等于1的数据，全部交给线程1处理
 * 余数等于2的数据，全部交给线程2处理
 * 余数等于3的数据，全部交给线程3处理
 * 余数等于4的数据，全部交给线程4处理
 * <p>
 * 线程中，就是计算获得的数据里面的 每个事件的发生人数，统计完后，写入一个输出文件
 * 每个线程各自输出一个结果文件
 **/
public class _05_综合练习_并发统计 {

    public static void main(String[] args) throws Exception {

        // 调用工具，生成数据文件
        DataGenerator.generateData();


        // 1.读文件，并将每行添加到bigList
        //List<String> bigList = FileUtils.readLines(new File("e:/data.txt"), "UTF-8");
        ArrayList<String> bigList = new ArrayList<>();
        FileInputStream fis = new FileInputStream("e:/data.txt");
        InputStreamReader inputStreamReader = new InputStreamReader(fis, "UTF-8");
        BufferedReader br = new BufferedReader(inputStreamReader);

        String line;
        while ((line = br.readLine()) != null) {
            if (!line.trim().isEmpty()) {
                bigList.add(line);
            }
        }

        br.close();
        System.out.println("数据已经全部加载到bigList , size ==> " + bigList.size());


        // 2. 然后，对上面的 bigList 进行拆分，拆分成 5个  smallList
        //    拆分的规则是： json数据中的 event_id.hashcode() % 5 得到余数， 余数相同的放到通过一个 smallList

        ArrayList<String> smallList_0 = new ArrayList<>();
        ArrayList<String> smallList_1 = new ArrayList<>();
        ArrayList<String> smallList_2 = new ArrayList<>();
        ArrayList<String> smallList_3 = new ArrayList<>();
        ArrayList<String> smallList_4 = new ArrayList<>();


        for (String lineJson : bigList) {
            // {"user_id":821,"event_id":"search","action_time":1707009932978}
            JSONObject jsonObject = JSON.parseObject(lineJson);
            String eventId = jsonObject.getString("event_id");

            // 字符串的 hashcode有 负数
            int ys = Math.abs(eventId.hashCode() % 5);
            switch (ys) {
                case 0:
                    smallList_0.add(lineJson);
                    break;
                case 1:
                    smallList_1.add(lineJson);
                    break;
                case 2:
                    smallList_2.add(lineJson);
                    break;
                case 3:
                    smallList_3.add(lineJson);
                    break;
                case 4:
                    smallList_4.add(lineJson);
                    break;
                default:
                    break;
            }

        }

        System.out.println(smallList_0.size());
        System.out.println(smallList_1.size());
        System.out.println(smallList_2.size());
        System.out.println(smallList_3.size());
        System.out.println(smallList_4.size());


        // 4. 启动 5 个线程，给每个线程一个 smallList
        new Thread(new CalculateTask(smallList_0,0)).start();
        new Thread(new CalculateTask(smallList_1,1)).start();
        new Thread(new CalculateTask(smallList_2,2)).start();
        new Thread(new CalculateTask(smallList_3,3)).start();
        new Thread(new CalculateTask(smallList_4,4)).start();




    }
}

class CalculateTask implements Runnable{
    List<String> smallList;
    int taskId;

    public CalculateTask(List<String> smallList,int taskId){
        this.taskId = taskId;
        this.smallList = smallList;
    }


    @Override
    public void run() {

        HashMap<String, Set<Integer>> map = new HashMap<>();

        for (String json : smallList) {
            JSONObject jsonObject = JSON.parseObject(json);

            // {"user_id":821,"event_id":"search","action_time":1707009932978}
            int userId = jsonObject.getIntValue("user_id");
            String eventId = jsonObject.getString("event_id");

            Set<Integer> userIdSet = map.get(eventId);
            if(userIdSet == null) {
                userIdSet = new HashSet<>();
            }
            // 添加本条数据的行为人
            userIdSet.add(userId);
            map.put(eventId,userIdSet);

        }

        /**
         * 输出结果
         */
        try {
            // 构造输出流
            BufferedWriter bufferedWriter = new BufferedWriter(new OutputStreamWriter(new FileOutputStream("e:/result-" + taskId + ".txt"), "UTF-8"));

            // 遍历结果map
            Set<Map.Entry<String, Set<Integer>>> entries = map.entrySet();
            for (Map.Entry<String, Set<Integer>> entry : entries) {
                String eventId = entry.getKey();
                Set<Integer> userIdSet = entry.getValue();

                bufferedWriter.write(eventId+","+userIdSet.size());
                bufferedWriter.newLine();
            }

            bufferedWriter.flush();
            bufferedWriter.close();

        }catch (Exception e){

        }

    }
}






class DataGenerator {
    public static void generateData() throws Exception {

        // 构造一个文件输出流
        FileOutputStream fos = new FileOutputStream("e:/data.txt");
        OutputStreamWriter outputStreamWriter = new OutputStreamWriter(fos, "UTF-8");
        BufferedWriter bufferedWriter = new BufferedWriter(outputStreamWriter);


        // 定义一个数据模板
        String template = "{\"user_id\":${1},\"event_id\":\"${2}\",\"action_time\":${3}}";
        String[] events = new String[]{"video_play", "video_pause", "ad_show", "ad_click", "search", "page_load", "article_share"};
        Random random = new Random();

        for (int i = 0; i < 20; i++) {
            // 生成数据值
            int user_id = random.nextInt(1000);
            String event_id = events[random.nextInt(events.length)];
            long action_time = System.currentTimeMillis();

            // 用生成的数据替换模板中的占位符
            String replace1 = template.replace("${1}", user_id + "");
            String replace2 = replace1.replace("${2}", event_id);
            String replace3 = replace2.replace("${3}", action_time + "");

            // 用输出流输出生成好的数据
            bufferedWriter.write(replace3);
            bufferedWriter.newLine();

        }


        bufferedWriter.flush();
        bufferedWriter.close();


    }
}

